package com.ttx.zookeeper.pubsub.curator;

import com.ttx.zookeeper.pubsub.curator.models.Instance;
import com.ttx.zookeeper.pubsub.curator.models.InstanceType;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.x.async.AsyncCuratorFramework;
import org.apache.curator.x.async.AsyncStage;
import org.apache.curator.x.async.modeled.ModeledFramework;
import org.apache.curator.x.async.modeled.cached.CachedModeledFramework;
import org.apache.curator.x.async.modeled.cached.ModeledCacheListener;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * fixme 尚未测试成功, 出现KeeperErrorCode = Unimplemented
 * @see Clients#builder(Class)
 *
 * @author TimFruit
 * @date 20-2-28 下午8:25
 */
public class PubSubSample {

    private static final AtomicLong nextId = new AtomicLong(1);

    private static final String[] hostnames = {"host1", "host2", "host3"};
    private static final Integer[] ports = {80, 443, 9999};


    public static void main(String[] args) throws InterruptedException {
        String connectString = "127.0.0.1:2181," +
                "127.0.0.1:2182" ;

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);
        client.start();


        AsyncCuratorFramework asyncClient = AsyncCuratorFramework.wrap(client);


        Instance instance = new Instance(nextId(), random(InstanceType.values()), random(hostnames), random(ports));
        System.out.println("Publishing 1 instance");
        ModeledFramework<Instance> resolvedClient = Clients.instanceClient.resolved(asyncClient, instance.getType());


        //1. 订阅
        Subscriber subscriber = new Subscriber(asyncClient);
        CachedModeledFramework<Instance> cmf = subscriber.startInstanceSubscriber(InstanceType.web);

        //2. 添加监听器 (收到信息进行处理)
        cmf.listenable().addListener(generalListener());


        //3. 发布
        Publisher publisher = new Publisher(asyncClient);


        ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
        // schedule the publisher task once a second
        executorService.scheduleAtFixedRate(() -> publishSomething(publisher), 1, 5, TimeUnit.SECONDS);

        TimeUnit.MINUTES.sleep(1);


    }


    private static void publishSomething(Publisher publisher) {
        // randomly do some publishing - either single items or lists of items in a transaction
        switch (ThreadLocalRandom.current().nextInt(1)) {
            case 0: {
                Instance instance = new Instance(nextId(), random(InstanceType.values()), random(hostnames), random(ports));
                System.out.println("Publishing 1 instance");
                publisher.publishInstance(instance);
                break;
            }

            case 1: {
                List<Instance> instances = IntStream.range(1, 10)
                        .mapToObj(__ -> new Instance(nextId(), random(InstanceType.values()), random(hostnames), random(ports)))
                        .collect(Collectors.toList());
                System.out.println(String.format("Publishing %d instances", instances.size()));
                publisher.publishInstances(instances);
                break;
            }
        }
    }


    private static <T> ModeledCacheListener<T> generalListener() {
        return (type, path, stat, model) -> System.out.println(String.format("Subscribed %s @ %s", model.getClass().getSimpleName(), path));
    }


    @SafeVarargs
    private static final <T> T random(T... tab) {
        int index = ThreadLocalRandom.current().nextInt(tab.length);
        return tab[index];
    }

    private static String nextId() {
        return Long.toString(nextId.getAndIncrement());
    }
}
